package rts

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"strings"
	"time"

	"github.com/google/uuid"

	"code.byted.org/bytertc/whiteboard_server/biz/api"
	"code.byted.org/bytertc/whiteboard_server/biz/api/pb/rts_pb"
	v1 "code.byted.org/bytertc/whiteboard_server/biz/api/pb/v1"
	"code.byted.org/bytertc/whiteboard_server/biz/api/pb/wb_pb"
	"code.byted.org/bytertc/whiteboard_server/biz/config"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/gateway"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/grpc/grpc_client"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/syncx"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/util"
	"code.byted.org/gopkg/env"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/middleware/hertz/byted"
	"code.byted.org/middleware/hertz/byted/discovery"
	hertzConfig "code.byted.org/middleware/hertz/pkg/common/config"
	"code.byted.org/middleware/hertz/pkg/protocol"
	"github.com/bytedance/sonic"
	"github.com/pkg/errors"
	"github.com/shirou/gopsutil/cpu"
	"github.com/shirou/gopsutil/disk"
	"github.com/shirou/gopsutil/mem"
	"google.golang.org/protobuf/proto"
)

const (
	PathHeartbeat    = "/whiteboard_scheduler/heartbeat"
	PathRoomExit     = "/whiteboard_scheduler/room_exit"
	PathInstanceExit = "/whiteboard_scheduler/instance_exit"
)

type Manager interface {
	//from scheduler
	JoinRoom(ctx context.Context, request *api.SchedulerJoinRoomRequest) (*api.SchedulerJoinRoomResponse, error)
	LeaveRoom(ctx context.Context, request *api.SchedulerLeaveRoomRequest) (*api.SchedulerLeaveRoomResponse, error)

	//server
	UpdateLastActiveTime(ctx context.Context, appID, roomID string)

	//from rts proxy
	RtsHeartbeat(ctx context.Context, appID string, roomIDs []string)
	RtsEvent(ctx context.Context, eventType wb_pb.EventType, appID, roomID, content string)

	//to rts_proxy
	SendUnicastMsg(ctx context.Context, appID, userID, data string) error
	SendBroadcastMsg(ctx context.Context, appID, roomID, data string) error

	//to rts_openapi
	ChangeServerConfig(ctx context.Context, appID, roomID, serverUID string, serverVersion int32) error

	//优雅退出
	InstanceExit()

	StartHeartbeat()
	StartMonitor()
}

type Engine struct {
	AppID          string
	ServerUID      string
	GrpcServerAddr string
	Rooms          map[string]*Room
}

type Room struct {
	AppID             string `json:"app_id"`
	RoomID            string `json:"room_id"`
	LastActiveTime    int64  `json:"last_active_time"`
	LastHeartbeatTime int64  `json:"last_heartbeat_time"`
}

type Event struct {
	AppID          string
	GrpcServerAddr string
	Success        bool
}

type manager struct {
	mLock          syncx.MultipleLock
	IP             string
	Port           string
	grpcServerAddr string
	engines        map[string]*Engine
	events         map[string]chan *Event
	connManager    *grpc_client.ConnectionManager
	rtsMsgClient   map[string]rts_pb.RtsProxy_RtsMsgClient
	rtcAPI         gateway.RtcAPI
	httpClient     *byted.Client
	heartbeat      bool
}

func NewManager(rtcAPI gateway.RtcAPI, connManager *grpc_client.ConnectionManager, httpClient *byted.Client, grpcServerAddr string) Manager {
	return &manager{
		mLock:          syncx.NewMultipleLock(),
		IP:             env.IPStr(),
		Port:           os.Getenv("PORT0"),
		engines:        make(map[string]*Engine),
		events:         make(map[string]chan *Event),
		grpcServerAddr: grpcServerAddr,
		rtcAPI:         rtcAPI,
		connManager:    connManager,
		rtsMsgClient:   make(map[string]rts_pb.RtsProxy_RtsMsgClient),
		httpClient:     httpClient,
		heartbeat:      true,
	}
}

func (m *manager) JoinRoom(ctx context.Context, request *api.SchedulerJoinRoomRequest) (*api.SchedulerJoinRoomResponse, error) {
	m.mLock.Lock(request.AppID)
	defer m.mLock.Unlock(request.AppID)
	if engine, ok := m.engines[request.AppID]; !ok || engine.ServerUID != request.ServerUID {
		m.engines[request.AppID] = nil
		m.events[request.AppID] = make(chan *Event, 100)

		token, err := m.rtcAPI.GenByteToken(ctx, request.AppID, request.RoomID, request.ServerUID)
		if err != nil {
			log.V1.CtxError(ctx, "JoinRoom gen token error:%v,app_id:%s,room_id:%s,user_id:%s", err, request.AppID, request.RoomID, request.ServerUID)
			return &api.SchedulerJoinRoomResponse{Success: false}, err
		}

		envNum := "0"
		if env.IsBoe() {
			envNum = "1"
		}
		if strings.HasSuffix(env.PSM(), "_test") {
			envNum = "2"
		}
		c := newCommand([]string{"--app_id=" + request.AppID, "--user_id=" + request.ServerUID, "--token=" + token, "--env=" + envNum, "--server_address=" + m.grpcServerAddr})
		go func(appID string, c *cmd) {
			defer util.CheckPanic()
			c.launch(context.Background())
			m.events[appID] <- &Event{
				AppID:   appID,
				Success: false,
			}
		}(request.AppID, c)

		select {
		case event := <-m.events[request.AppID]:
			if event.Success {
				m.engines[request.AppID] = &Engine{
					AppID:          request.AppID,
					ServerUID:      request.ServerUID,
					GrpcServerAddr: event.GrpcServerAddr,
					Rooms:          make(map[string]*Room),
				}
			} else {
				log.V1.CtxError(ctx, "JoinRoom create event not Success,app_id:%s,room_id:%s,user_id:%s", request.AppID, request.RoomID, request.ServerUID)
				return &api.SchedulerJoinRoomResponse{Success: false}, errors.New("create event not Success")
			}
		case <-time.After(5 * time.Second):
			log.V1.CtxError(ctx, "JoinRoom create event timeout,app_id:%s,room_id:%s,user_id:%s", request.AppID, request.RoomID, request.ServerUID)
			return &api.SchedulerJoinRoomResponse{Success: false}, errors.New("create event timeout")
		}
	}
	if _, ok := m.engines[request.AppID].Rooms[request.RoomID]; !ok {
		rtsProxyClient := m.grpcClient(ctx, request.AppID)
		if rtsProxyClient == nil {
			return &api.SchedulerJoinRoomResponse{Success: false}, errors.New("rts proxy client nil")
		}
		resp, err := rtsProxyClient.JoinRoom(ctx, &rts_pb.JoinRoomRequest{
			AppId:  request.AppID,
			RoomId: request.RoomID,
		})
		if err != nil {
			log.V1.CtxError(ctx, "JoinRoom join room error:%v,app_id:%s,room_id:%s", err, request.AppID, request.RoomID)
			return &api.SchedulerJoinRoomResponse{Success: false}, err
		}
		if resp.Status != 0 {
			log.V1.CtxError(ctx, "JoinRoom join room error,app_id:%s,,room_id:%s,status:%v,err_msg:%s", request.AppID, request.RoomID, resp.Status, resp.ErrMsg)
			return &api.SchedulerJoinRoomResponse{Success: false}, errors.New("resp status not Success")
		}
		m.engines[request.AppID].Rooms[request.RoomID] = &Room{
			AppID:             request.AppID,
			RoomID:            request.RoomID,
			LastActiveTime:    time.Now().Unix(),
			LastHeartbeatTime: time.Now().Unix(),
		}
	}
	return &api.SchedulerJoinRoomResponse{Success: true}, nil
}

func (m *manager) LeaveRoom(ctx context.Context, request *api.SchedulerLeaveRoomRequest) (*api.SchedulerLeaveRoomResponse, error) {
	m.mLock.Lock(request.AppID)
	defer m.mLock.Unlock(request.AppID)
	if request.IsMigrate {
		err := m.ChangeServerConfig(ctx, request.AppID, request.RoomID, request.ServerUID, request.ServerVersion)
		if err != nil {
			log.V1.CtxError(ctx, "LeaveRoom ChangeServerConfig error:%v", err)
		}
	}
	if engine, ok := m.engines[request.AppID]; ok {
		if _, ook := engine.Rooms[request.RoomID]; ook {
			rtsProxyClient := m.grpcClient(ctx, request.AppID)
			if rtsProxyClient == nil {
				return &api.SchedulerLeaveRoomResponse{}, nil
			}
			resp, err := rtsProxyClient.LeaveRoom(ctx, &rts_pb.LeaveRoomRequest{
				AppId:  request.AppID,
				RoomId: request.RoomID,
			})
			if err != nil {
				log.V1.CtxError(ctx, "leave room error:%v,app_id:%s,room_id:%s", err, request.AppID, request.RoomID)
				return &api.SchedulerLeaveRoomResponse{}, nil
			}
			if resp.Status != 0 {
				log.V1.CtxError(ctx, "leave room error,app_id:%s,,room_id:%s,status:%v,err_msg:%s", request.AppID, request.RoomID, resp.Status, resp.ErrMsg)
				return &api.SchedulerLeaveRoomResponse{}, nil
			}
		}
		delete(engine.Rooms, request.RoomID)
	}
	return &api.SchedulerLeaveRoomResponse{}, nil
}

func (m *manager) UpdateLastActiveTime(ctx context.Context, appID, roomID string) {
	defer util.CheckPanic()
	if engine, ok := m.engines[appID]; ok {
		if room, ook := engine.Rooms[roomID]; ook {
			room.LastActiveTime = time.Now().Unix()
		}
	}
}

func (m *manager) RtsHeartbeat(ctx context.Context, appID string, roomIDs []string) {
	defer util.CheckPanic()
	if engine, ok := m.engines[appID]; ok {
		for _, roomID := range roomIDs {
			if room, ook := engine.Rooms[roomID]; ook {
				room.LastHeartbeatTime = time.Now().Unix()
			}
		}
	}
}

func (m *manager) RtsEvent(ctx context.Context, eventType wb_pb.EventType, appID, roomID, content string) {
	defer util.CheckPanic()
	switch eventType {
	case wb_pb.EventType_kEngineCreateSuccess:
		if _, ok := m.events[appID]; ok {
			addr := "unix:" + content
			_, err := m.connManager.Put(appID, addr)
			if err != nil {
				log.V1.CtxError(ctx, "connection build error:%v", err)
				return
			}
			m.events[appID] <- &Event{
				AppID:          appID,
				GrpcServerAddr: addr,
				Success:        true,
			}
		}
	case wb_pb.EventType_kEngineAbnormal:
		if _, ok := m.events[appID]; ok {
			m.events[appID] <- &Event{AppID: appID, Success: false}
		}
	case wb_pb.EventType_kRoomAbnormal:
		roomExitRequest := &RoomExitRequest{
			AppID:  appID,
			RoomID: roomID,
		}
		_, err := m.postScheduler(ctx, PathRoomExit, roomExitRequest)
		if err != nil {
			log.V1.CtxError(ctx, "leave room error:%v", err)
		}
		delete(m.engines[appID].Rooms, roomID)
	case wb_pb.EventType_kJoinRoomSuccess:
		log.V1.CtxInfo(ctx, "join room Success,app_id:%s,room_id:%s,content:%s", appID, roomID, content)
	case wb_pb.EventType_kLeaveRoomSuccess:
		log.V1.CtxInfo(ctx, "leave room Success,app_id:%s,room_id:%s,content:%s", appID, roomID, content)
	}
}

func (m *manager) SendUnicastMsg(ctx context.Context, appID, userID, data string) (err error) {
	client, ok := m.rtsMsgClient[appID]
	if !ok {
		rtsProxyClient := m.grpcClient(ctx, appID)
		if rtsProxyClient == nil {
			return errors.New("rts proxy client nil")
		}
		client, err = rtsProxyClient.RtsMsg(ctx)
		if err != nil {
			return err
		}
		m.rtsMsgClient[appID] = client
	}
	err = client.Send(&rts_pb.RtsMsgRequest{
		AppId:       appID,
		UserId:      userID,
		IsBroadcast: false,
		Data:        []byte(data),
	})
	if err != nil {
		delete(m.rtsMsgClient, appID)
		return err
	}
	return nil
}

func (m *manager) SendBroadcastMsg(ctx context.Context, appID, roomID, data string) (err error) {
	client, ok := m.rtsMsgClient[appID]
	if !ok {
		rtsProxyClient := m.grpcClient(ctx, appID)
		if rtsProxyClient == nil {
			return errors.New("rts proxy client nil")
		}
		client, err = rtsProxyClient.RtsMsg(ctx)
		if err != nil {
			return err
		}
		m.rtsMsgClient[appID] = client
	}
	err = client.Send(&rts_pb.RtsMsgRequest{
		AppId:       appID,
		RoomId:      roomID,
		IsBroadcast: true,
		Data:        []byte(data),
	})
	if err != nil {
		delete(m.rtsMsgClient, appID)
		return err
	}
	return nil
}

func (m *manager) ChangeServerConfig(ctx context.Context, appID, roomID, serverUID string, serverVersion int32) error {
	log.V1.CtxInfo(ctx, "ChangeServerConfig,appID:%s,roomID:%s,serverUID:%s", appID, roomID, serverUID)
	msg := &v1.ChangeServerConfigMessage{
		ServerVersion: serverVersion,
		ServerUid:     serverUID,
	}
	data, err := proto.Marshal(msg)
	if err != nil {
		log.V1.CtxError(ctx, "ChangeServerUid Marshal error:%v", err)
		return errors.WithStack(err)
	}
	resp := &v1.OnWireResponse{
		Operator: "server",
		Command:  v1.Command_kChangeServerConfig,
		Data:     data,
		RoomId:   &roomID,
	}

	data, err = proto.Marshal(resp)
	if err != nil {
		log.V1.CtxError(ctx, "ChangeServerUid Marshal error:%v", err)
		return errors.WithStack(err)
	}

	err, errCode := m.rtcAPI.ToRoom(ctx, appID, roomID, EncodeBase64(data), "unknown")
	if err != nil {
		log.V1.CtxError(ctx, "ChangeServerUid ToRoom error:%v,err_code:%v", err, errCode)
		return err
	}
	return nil
}

func (m *manager) InstanceExit() {
	ctx := util.EnsureID(context.Background())
	log.V1.CtxInfo(ctx, "InstanceExit")
	m.heartbeat = false
	instanceExitRequest := &InstanceExitRequest{
		IP:   m.IP,
		Port: m.Port,
	}
	_, err := m.postScheduler(ctx, PathInstanceExit, instanceExitRequest)
	if err != nil {
		log.V1.CtxError(ctx, "instanceExit error:%v", err)
	}
	time.Sleep(20 * time.Second)
	log.V1.CtxInfo(ctx, "InstanceExit 20s")
	time.Sleep(30 * time.Second)
}

func (m *manager) StartHeartbeat() {
	ctx := util.EnsureID(context.Background())
	go func(m *manager) {
		defer util.CheckPanic()
		ticker := time.NewTicker(5 * time.Second)
		for range ticker.C {
			if !m.heartbeat {
				return
			}
			heartbeatRequest := &HeartBeatRequest{
				IP:      m.IP,
				Port:    m.Port,
				Engines: make(map[string]*SEngine),
				Cpu:     getCpuPercent(),
				Mem:     getMemPercent(),
			}
			for _, engine := range m.engines {
				heartbeatRequest.Engines[engine.AppID] = &SEngine{
					AppID:     engine.AppID,
					ServerUID: engine.ServerUID,
					Rooms:     make(map[string]*SRoom),
				}
				for _, room := range engine.Rooms {
					heartbeatRequest.Engines[engine.AppID].Rooms[room.RoomID] = &SRoom{
						AppID:          room.AppID,
						RoomID:         room.RoomID,
						LastActiveTime: room.LastActiveTime,
					}
				}
			}

			resp, err := m.postScheduler(ctx, PathHeartbeat, heartbeatRequest)
			if err != nil {
				log.V1.CtxError(ctx, "heartbeat error:%v", err)
				continue
			}
			heartbeatResponse := &HeartBeatResponse{}
			err = json.Unmarshal(resp, heartbeatResponse)
			if err != nil {
				log.V1.CtxError(ctx, "heartbeat response unmarshal error:%v", err)
				continue
			}
		}
	}(m)
}

func (m *manager) StartMonitor() {
	go func() {
		defer util.CheckPanic()
		ticker := time.NewTicker(5 * time.Second)
		for range ticker.C {
			ctx := util.EnsureID(context.Background())
			cRooms := make(map[string][]string)
			for _, engine := range m.engines {
				cRooms[engine.AppID] = make([]string, 0)
				for _, room := range engine.Rooms {
					if time.Since(time.Unix(room.LastActiveTime, 0)) > 12*time.Hour {
						log.V1.CtxInfo(ctx, "Monitor NotActive app_id:%s,room_id:%s", room.AppID, room.RoomID)
						cRooms[engine.AppID] = append(cRooms[engine.AppID], room.RoomID)
					} else if time.Since(time.Unix(room.LastHeartbeatTime, 0)) > 25*time.Second {
						log.V1.CtxInfo(ctx, "Monitor NotHeartbeat app_id:%s,room_id:%s", room.AppID, room.RoomID)
						cRooms[engine.AppID] = append(cRooms[engine.AppID], room.RoomID)
					}
				}
			}
			for appID, rooms := range cRooms {
				engineNormal := true
				for _, roomID := range rooms {
					delete(m.engines[appID].Rooms, roomID)
					rtsProxyClient := m.grpcClient(ctx, appID)
					if rtsProxyClient == nil {
						engineNormal = false
						break
					}
					resp, err := rtsProxyClient.LeaveRoom(ctx, &rts_pb.LeaveRoomRequest{
						AppId:  appID,
						RoomId: roomID,
					})
					if err != nil {
						log.V1.CtxError(ctx, "leave room error:%v,app_id:%s,room_id:%s", err, appID, roomID)
						continue
					}
					if resp.Status != 0 {
						log.V1.CtxError(ctx, "leave room error,app_id:%s,,room_id:%s,status:%v,err_msg:%s", appID, roomID, resp.Status, resp.ErrMsg)
						continue
					}
				}

				if engineNormal {
					if len(m.engines[appID].Rooms) == 0 {
						m.FinishRts(ctx, appID)
					}
				}
			}
		}
	}()
}
func (m *manager) FinishRts(ctx context.Context, appID string) {
	log.V1.CtxInfo(ctx, "Finish Rts")
	rtsProxyClient := m.grpcClient(ctx, appID)
	if rtsProxyClient == nil {
		log.V1.CtxError(ctx, "get rts client failed")
		return
	}
	resp, err := rtsProxyClient.Finish(ctx, &rts_pb.FinishRequest{
		AppId: appID,
	})
	delete(m.engines, appID)
	delete(m.rtsMsgClient, appID)
	m.connManager.Delete(appID)

	if err != nil {
		log.V1.CtxError(ctx, "finish error:%v,app_id:%s", err, appID)
		return
	}
	if resp.Status != 0 {
		log.V1.CtxError(ctx, "finish error,app_id:%s,,status:%v,err_msg:%s", appID, resp.Status, resp.ErrMsg)
		return
	}
}

func (m *manager) postScheduler(ctx context.Context, path string, body interface{}) ([]byte, error) {
	data, _ := sonic.Marshal(body)
	query := url.Values{}

	req := &protocol.Request{}
	req.SetOptions(discovery.WithDestinationCluster(env.Cluster()))
	req.SetMethod(http.MethodPost)
	req.SetBody(data)
	req.SetRequestURI("http://" + config.GetConfig().SchedulerPsm + path)

	req.SetOptions(hertzConfig.WithSD(true))
	req.SetQueryString(query.Encode())
	req.SetHeader("Content-Type", "application/json")

	resp := &protocol.Response{}
	err := m.httpClient.Do(ctx, req, resp)
	if err != nil {
		return nil, errors.WithStack(err)
	}
	if resp.StatusCode() != http.StatusOK {
		return nil, errors.New("http status is not ok")
	}
	return resp.Body(), nil
}

func (m *manager) grpcClient(ctx context.Context, appID string) rts_pb.RtsProxyClient {
	conn, err := m.connManager.Get(appID)
	if err != nil {
		conn, err = m.connManager.Put(appID, m.engines[appID].GrpcServerAddr)
		if err != nil {
			log.V1.CtxError(ctx, "get conn error:%v,app_id:%s", err, appID)
			delete(m.engines, appID)
			delete(m.rtsMsgClient, appID)
			m.connManager.Delete(appID)
			return nil
		}
	}
	return rts_pb.NewRtsProxyClient(conn)
}

func getCpuPercent() float64 {
	percent, _ := cpu.Percent(time.Second, false)
	return percent[0]
}

func getMemPercent() float64 {
	mem.VirtualMemory()
	memInfo, _ := mem.VirtualMemory()
	return memInfo.UsedPercent
}

func getDiskPercent() float64 {
	parts, _ := disk.Partitions(true)
	diskInfo, _ := disk.Usage(parts[0].Mountpoint)
	return diskInfo.UsedPercent
}

type cmd struct {
	cmd     *exec.Cmd
	cmdName string
	args    []string
}

func newCommand(args []string) *cmd {
	return &cmd{
		cmdName: "/opt/tiger/whiteboard_rts_proxy/bin/bytertc.server.whiteboard_rts_proxy",
		args:    args,
	}
}

func (c *cmd) launch(ctx context.Context) {
	log.V1.CtxInfo(ctx, "cmd name :%v,cmd args:%+v", c.cmdName, c.args)
	file, err := os.Create(fmt.Sprintf("/tmp/whiteboard/%s", uuid.NewString()))
	if err != nil {
		log.V1.CtxError(ctx, "cmd launch create file error:%v", err)
	}
	c.cmd = exec.Command(c.cmdName, c.args...)
	c.cmd.Stdout = file
	c.cmd.Stderr = file

	err = c.cmd.Start()
	if err != nil {
		log.V1.CtxError(ctx, "start command error:%v,args:%v", err, c.args)
		return
	}

	err = c.cmd.Wait()
	if err != nil {
		log.V1.CtxError(ctx, "wait  command error:%v,args:%v", err, c.args)
		return
	}
}
